package lowlevel;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author:RenPu
 * @Date: 2020/3/27 15:25
 * @Version: 1.0
 * @description:单词计数
 */
public class WordCountProcesser implements Processor<String,String[]> {

    private ProcessorContext context=null;

    //创建一个暂存的map集合

    private Map<String,Long> wordCount=null;

    @Override
    public void init(ProcessorContext context) {

        //初始化对象
        this.wordCount=new HashMap(20);
        this.context=context;
        this.context.schedule(Duration.ofSeconds(1), PunctuationType.STREAM_TIME,(timeStamp->{
            wordCount.forEach((word,num)->{
                context.forward(word,num);
            });
        }));

    }

    @Override
    public void process(String s, String[] words) {

        for (String word : words) {

            //获取指定的key对应的value,如果获取v不存在 num=0L;
            Long num= wordCount.getOrDefault(word, 0L);

            num++;

            //将计数累加后的值，以单词作为key，value进行存储
            wordCount.put(word,num);
        }

        //标识处理结果
        context.commit();

    }

    @Override
    public void close() {

    }
}
